use optional_struct::optional_struct;
use optional_struct::Applyable;
use serde::{Deserialize, Serialize};
use std::{
    collections::HashMap,
    sync::Arc,
    time::{Duration, SystemTime},
};

use ekho::kcp::Config;
use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};

use futures_util::{AsyncReadExt, AsyncWriteExt, SinkExt, StreamExt};
use tokio::sync::RwLock;

use webrtc_util::Conn;

use async_trait::async_trait;

use webrtc_ice::{
    agent::{agent_config::AgentConfig, Agent},
    candidate::{candidate_base::unmarshal_candidate, Candidate},
    mdns::MulticastDnsMode,
    network_type::NetworkType,
    state::ConnectionState,
    url::Url,
};

use crate::imp::tonicrpc::ekhokcp::kcp_imp::EkhoKcpImp;
use crate::imp::tonicrpc::ekhokcp::kcp_imp::WebRtcConnToAsyncKcpTunnel;
use crate::imp::tonicrpc::kcptunnel::kcp::KcpTunnelImp;
use crate::imp::tonicrpc::tonicserver::close_tunnel_connect;
use crate::imp::tonicrpc::tonicserver::if_no_kcp_set_err;
use crate::imp::tonicrpc::tonictunnel::TonicTunnelCallback;

use crate::utils::ip_filter;

pub const TUNNEL_ID_SENSOR_END_FIX: &'static str = "--SENSOR--";
pub const TUNNEL_ID_BYTEBUF_END_FIX: &'static str = "--BYTES--";
pub const TUNNEL_ID_FILE_END_FIX: &'static str = "--FILE--";
pub const TUNNEL_ID_HTTP_END_FIX: &'static str = "--HTTP--";

// 连接通道，tunnel_id没有后缀
pub fn is_linked_tunnel(tunnel_id: String) -> bool {
    !is_sensor_tunnel(tunnel_id.clone())
        && !is_file_tunnel(tunnel_id.clone())
        && !is_bb_tunnel(tunnel_id.clone())
        && !is_http_tunnel(tunnel_id.clone())
}

// sensor通道,tunnel_id附带有TUNNEL_ID_SENSOR_END_FIX后缀
pub fn is_sensor_tunnel(tunnel_id: String) -> bool {
    tunnel_id.ends_with(TUNNEL_ID_SENSOR_END_FIX)
}

// 文件通道，tunnel_id附带有TUNNEL_ID_FILE_END_FIX后缀
pub fn is_file_tunnel(tunnel_id: String) -> bool {
    tunnel_id.ends_with(TUNNEL_ID_FILE_END_FIX)
}

// 字节数组通道，tunnel_id附带有TUNNEL_ID_BYTEBUF_END_FIX后缀
pub fn is_bb_tunnel(tunnel_id: String) -> bool {
    tunnel_id.ends_with(TUNNEL_ID_BYTEBUF_END_FIX)
}

// HTTP通道，tunnel_id附带有TUNNEL_ID_HTTP_END_FIX后缀
pub fn is_http_tunnel(tunnel_id: String) -> bool {
    tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX)
}

fn default_u64() -> u64 {
    300
}

fn default_string() -> String {
    "".to_string()
}

#[derive(PartialEq, Debug, Default, Serialize, Deserialize, Clone)]
pub struct URIInfo {
    #[serde(rename = "username", default = "default_string")]
    pub username: String,
    #[serde(rename = "uri")]
    pub uri: String,
    #[serde(rename = "ttl", default = "default_u64")]
    pub ttl: u64,
    #[serde(rename = "password", default = "default_string")]
    pub password: String,
}

/// URIInfo
impl URIInfo {
    // from custom uriinfo to webrtc url.
    fn to_webrtc_url(&self) -> Result<(Url, Url), anyhow::Error> {
        let URIInfo {
            username,
            uri,
            ttl: _,
            password,
        } = self;
        log::info!("url parse_url url:{:?}", uri);

        let has_prefix = if uri.starts_with("turn:") || uri.starts_with("stun:") {
            true
        } else {
            false
        };
        let turn_url = if has_prefix && uri.starts_with("turn:") {
            uri.to_owned()
        } else if has_prefix {
            let replace_uri = str::replace(&uri, "stun:", "turn:");
            replace_uri
        } else {
            format!("turn:{}", uri).to_owned()
        };

        let stun_url = str::replace(&turn_url.clone(), "turn:", "stun:");
        let mut stun_webrtc_url = Url::parse_url(&stun_url)?;
        stun_webrtc_url.password = password.to_owned();
        stun_webrtc_url.username = username.to_owned();

        let mut turn_webrtc_url = Url::parse_url(&turn_url)?;
        turn_webrtc_url.password = password.to_owned();
        turn_webrtc_url.username = username.to_owned();
        log::info!(
            "to_url stun url: {:?} , turn url:{:?} ",
            stun_url,
            turn_webrtc_url
        );
        Ok((stun_webrtc_url, turn_webrtc_url))
    }
}

#[optional_struct]
#[derive(Debug, Serialize, Deserialize, Clone)]
// tunnel config
pub struct TunnelConfig {
    pub config_store_path: String,
    pub client_id: String,
    pub turn_uri: URIInfo,

    pub download_base_path: String,
    pub enable_http_callback: bool,
}

impl TunnelConfig {
    fn default() -> Self {
        Self {
            config_store_path: "./".to_owned(),
            client_id: "default.device".to_owned(),
            turn_uri: URIInfo {
                username: "default.turn_usr".to_owned(),
                password: "default.turn_pwd".to_owned(),
                uri: "turn:127.0.0.1:3478".to_owned(), // can not prefix //
                ttl: 0,
            },
            download_base_path: "/".to_owned(),
            enable_http_callback: false,
        }
    }
}

// connection for one pair!
#[derive(Clone)]
pub struct TunnelConn {
    // ice_agent
    pub ice_agent: Arc<Agent>,
    // connection for ice
    pub conn: Arc<dyn Conn + Send + Sync>,
    // remote client id
    pub remote_client_id: String,
    // local client id
    pub local_client_id: String,
    // success tunnel id
    pub tunnel_id: String,
    // mpsc cancel tx
    pub cancel_tx: tokio::sync::mpsc::Sender<()>,
    // who is controlling
    pub is_controlling: bool,
    // buffer to ekho
    pub ekho_config: Config,
}

impl TunnelConn {
    pub async fn to_kcp_tunnel(self) -> Result<KcpTunnelImp, anyhow::Error> {
        let TunnelConn {
            ice_agent,
            conn,
            remote_client_id,
            local_client_id,
            tunnel_id,
            cancel_tx,
            is_controlling,
            ekho_config,
        } = self;

        let stream =
            EkhoKcpImp::to_asnyc_kcp_tunnel(conn, cancel_tx.clone(), &tunnel_id, ekho_config)
                .await?;

        Ok(KcpTunnelImp::new(
            stream,
            ice_agent,
            remote_client_id,
            local_client_id,
            tunnel_id,
            cancel_tx,
            is_controlling,
        ))
    }
}

impl std::fmt::Debug for TunnelConn {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TunnelConn")
            .field("remote_client_id", &self.remote_client_id)
            .field("local_client_id", &self.local_client_id)
            .field("tunnel_id", &self.tunnel_id)
            .field("cancel_tx", &self.cancel_tx)
            .field("is_controlling", &self.cancel_tx)
            .field("ekho_config", &self.ekho_config)
            .finish()
    }
}

#[async_trait]
pub trait TunnelClientCreator: Clone {
    // request connect remote_client_id with tunnel_id
    async fn request_connect(
        &mut self,
        remote_client_id: &str,
        tunnel_id: &str,
    ) -> Result<String, anyhow::Error>;
    // disconnect tunnel_id
    fn disconnect(&mut self, tunnel_id: &str) -> Result<(), anyhow::Error>;
    // stop tonic server
    fn stop_tonic_server(&mut self) -> Result<(), anyhow::Error>;
    // gather candidates to get the candidates
    fn gather_candidates(&mut self) -> Result<(), anyhow::Error>;
}

#[derive(Debug, Serialize, Deserialize)]
pub struct IceAuth {
    // ufrag
    pub ufrag: String,
    // pwd
    pub pwd: String,
    pub auth_local_client_id: String,
    pub auth_remote_client_id: String,
    // kcp config
    pub ekho_config: Option<Config>,
    // tunnel id
    pub tunnel_id: String,
    // is controlling
    pub is_controlling: bool,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct IceCandidate {
    // canidate from ice protocol
    pub candidate: String,
    // candidate local client id
    pub can_local_client_id: String,
    // candidate remote client id
    pub can_remote_client_id: String,
    // is controlling
    pub is_controlling: bool,
    // unique tunnel id
    pub tunnel_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TunnelStopKind {
    // received remote client id close cmd
    RemoteClose,
    // received local client close cmd
    LocalClose,
    // received signal send fail
    SignalSendFail,
    /*   // received target off line
    * there are not need target offlient error!only need send fail.
    TargetOffline,*/
    // connect timeout
    ConnectTimout,
    // ice fail
    Failed,
}

impl std::fmt::Display for TunnelStopKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match *self {
            Self::RemoteClose => "RemoteClose",
            Self::LocalClose => "LocalClose",
            Self::SignalSendFail => "SignalSendFail",
            // Self::TargetOffline => "TargetOffline",
            Self::ConnectTimout => "ConnectTimout",
            Self::Failed => "Failed",
        };
        write!(f, "{}", s)
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub enum TunnelSignalIceMsg {
    // received remote auth command
    RemoteAuth(IceAuth),
    // received remote candidate command
    RemoteCandidate(IceCandidate),
    // received gather
    GatherCandidates {
        tunnel_id: String,
        gc_local_client_id: String,
        gc_remote_client_id: String,
    },
    // received stop
    Stop {
        tunnel_id: String,
        st_local_client_id: String,
        st_remote_client_id: String,
        kind: TunnelStopKind,
    },
}

#[derive(Debug)]
pub struct IceCreateMsg {
    pub is_controlling: bool,
    pub tunnel_id: String,
    pub remote_client_id: String,
    pub receive_rx: UnboundedReceiver<TunnelSignalIceMsg>,
}

#[derive(Debug)]
pub enum IceCommandMsg {
    // received ice create command
    Create(IceCreateMsg),
    // received disconnect tunnel command
    DisconnectTunnel(String),
    // received gather candidate command
    GatherCandidates,
    // received stop service command
    StopService,
}

#[derive(Debug, Clone)]
pub struct TunnelCreatorImp {
    // local client id,
    pub local_client_id: String,
    // sender_tx to send some ice creator message
    pub sender_tx: UnboundedSender<(String, TunnelSignalIceMsg)>,
    // commander to send some ice command message
    pub commander: UnboundedSender<IceCommandMsg>,
    // singal map,key = tunnel_id,value = tuple (IceMsg,remote_client_id,connectionstate)
    pub signal_channel_map: Arc<
        RwLock<HashMap<String, (UnboundedSender<TunnelSignalIceMsg>, String, ConnectionState)>>,
    >,
}

#[derive(Debug, Clone)]
pub struct TunnelStatus {
    pub tunnel_id: String,
    pub sa_local_client_id: String,
    pub sa_remote_client_id: String,
    pub state: ConnectionState,
    pub msg: String,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TunnelPair {
    pub tunnel_id: String,
    pub pair_local_client_id: String,
    pub pair_remote_client_id: String,
    pub remote_pair: String,
    pub mine_pair: String,
}

// fix lan ip access!
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum IpInLanType {
    SURE,
    MAYBE,
    NOT,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TunnelPairWithIp {
    pub mine_ip: Option<String>,
    pub remote_ip: Option<String>,
    pub ip_in_lan_type: IpInLanType,
    pub pair: TunnelPair,
}

impl TunnelPairWithIp {
    pub fn new(pair: TunnelPair) -> Self {
        let TunnelPair {
            tunnel_id: _,
            pair_local_client_id: _,
            pair_remote_client_id: _,
            remote_pair,
            mine_pair,
        } = pair.clone();

        let remote_ip = ip_filter::filter_ip_from(remote_pair.clone().to_owned());
        let mine_ip = ip_filter::filter_ip_from(mine_pair.to_owned());

        // first conditional, mine_pair must be "host" type
        // and remote_pair is "host" or "prflx" type
        // "host" type:
        //      10.X.X.X是私有地址，范围从10.0.0.0-10.255.255.255
        //      私有地址范围：172.16.0.0—172.31.255.255
        //      私有地址：192.168.X.X，范围从192.168.0.0-192.168.255.255

        let mut lan_type = IpInLanType::NOT;

        if mine_pair.contains("host") && remote_pair.contains("host") {
            lan_type = IpInLanType::SURE;
        } else if mine_pair.contains("host") && !remote_pair.contains("relay") {
            lan_type = IpInLanType::MAYBE;
        } else if remote_pair.contains("host") && !mine_pair.contains("relay") {
            lan_type = IpInLanType::MAYBE;
        } else {
            lan_type = IpInLanType::NOT;
        }

        TunnelPairWithIp {
            mine_ip: mine_ip,
            remote_ip: remote_ip,
            ip_in_lan_type: lan_type,
            pair: pair.to_owned(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TonicServerStopKind {
    ManualClose,
    ErrorClose,
}

impl std::fmt::Display for TonicServerStopKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match *self {
            Self::ManualClose => "ManualClose",
            Self::ErrorClose => "ErrorClose",
        };
        write!(f, "{}", s)
    }
}

#[derive(Clone, Debug)]
pub enum TunnelConnectMsg {
    // received on tunnel service created command
    OnTonicServerCreated(TunnelCreatorImp),
    // received on tunnel connect ready command
    OnTunnelConnectReady(TunnelConn),
    // received on tunnel state channge command
    OnTunnelStateChange(TunnelStatus),
    // received on tunnel pair change command
    OnTunnelPairChange(TunnelPair),
    // received on tunnel connect destroy command
    OnTunnelConnectDestroy {
        tunnel_id: String,
        local_client_id: String,
        remote_client_id: String,
    },
    // received on tunnel service drop command
    OnTonicServerDroped(TonicServerStopKind),
}

impl TunnelConnectMsg {
    pub fn on_connect_callback(
        self,
        callback: &mut Box<dyn TonicTunnelCallback>,
        enable_http_callback: bool,
    ) {
        match self {
            TunnelConnectMsg::OnTonicServerCreated(msg) => {
                callback.on_tonic_server_created(msg);
            }
            TunnelConnectMsg::OnTunnelConnectReady(msg) => {
                if !msg.tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX) || enable_http_callback {
                    callback.on_tunnel_connect_ready(msg);
                }
            }
            TunnelConnectMsg::OnTunnelStateChange(msg) => {
                if !msg.tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX) || enable_http_callback {
                    callback.on_tunnel_state_changed(msg);
                }
            }
            TunnelConnectMsg::OnTunnelPairChange(msg) => {
                if !msg.tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX) || enable_http_callback {
                    callback.on_tunnel_pair_changed(msg);
                }
            }
            TunnelConnectMsg::OnTunnelConnectDestroy {
                tunnel_id,
                local_client_id,
                remote_client_id,
            } => {
                if !tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX) || enable_http_callback {
                    callback.on_tunnel_connect_destroy(
                        tunnel_id,
                        local_client_id,
                        remote_client_id,
                    );
                }
            }
            TunnelConnectMsg::OnTonicServerDroped(msg) => {
                callback.on_tonic_server_droped(msg);
            }
        }
    }
}

fn candidate_to_json(m: Arc<dyn Candidate + Send + Sync>) -> String {
    let related_address = if let Some(related_address) = m.related_address() {
        format!("{}:{}", related_address.address, related_address.port,)
    } else {
        "".to_owned()
    };

    let obj = serde_json::json!({
        "foundation": m.foundation(),
        "component": m.component(),
        "network_type": m.network_type().network_short(),
        "priority": m.priority(),
        "address": m.address(),
        "port": m.port(),
        "candidate_type": m.candidate_type().to_string(),
        "related_address": related_address,
    });
    obj.to_string()
}

unsafe impl Send for TunnelCreatorImp {}
unsafe impl Sync for TunnelCreatorImp {}
impl TunnelCreatorImp {
    pub async fn async_new_tunnel_connect(
        // client: &reqwest::Client,
        is_controlling: bool,   // set is controlling
        local_client_id: &str,  // set local client id
        remote_client_id: &str, // set remote client id
        my_tunnel_id: &str,     // set my tunnel id
        mut receive_rx: UnboundedReceiver<TunnelSignalIceMsg>, // set ice message receiver
        mut sender_tx: UnboundedSender<(String, TunnelSignalIceMsg)>, // set ice message sender
        mut callback_tx: UnboundedSender<TunnelConnectMsg>, // set connector message call back
        options: &TunnelConfig,
    ) -> Result<(), anyhow::Error> {
        //take turn uri
        let TunnelConfig { turn_uri, .. } = options;
        let start_time = SystemTime::now();

        let config_uris = match turn_uri.to_webrtc_url() {
            Ok((stun_webrtc_uri, turn_webrtc_uri)) => (stun_webrtc_uri, turn_webrtc_uri),
            Err(error) => {
                anyhow::bail!(error);
            }
        };

        // uris.iter_mut().for_each(|item| {
        //     item.host = "159.75.233.101".to_owned();
        //     item.port = 43478;
        // });
        let urls = vec![config_uris.0, config_uris.1];
        // create ice_agent
        let ice_agent = Arc::new(
            Agent::new(AgentConfig {
                urls,
                multicast_dns_mode: MulticastDnsMode::Disabled,
                network_types: vec![
                    NetworkType::Udp4,
                    // NetworkType::Udp6,
                    // NetworkType::Tcp4,
                    // NetworkType::Tcp6,
                ],
                disconnected_timeout: Some(Duration::from_secs(0)),
                keepalive_interval: Some(Duration::from_secs(5)),
                ..Default::default()
            })
            .await?,
        );
        // agent internal user credentials generatation.
        let (local_ufrag, local_pwd) = ice_agent.get_local_user_credentials().await;

        log::info!(
            "new_tunnel_connect ice_agent create,local_ufrag = {},local_pwd={}",
            local_ufrag,
            local_pwd
        );

        let mut custom_kcp_config = Config::default();

        if my_tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX)
            || my_tunnel_id.ends_with(TUNNEL_ID_BYTEBUF_END_FIX)
        {
            // performance speed up for http sender
            custom_kcp_config.interval = 5;
            custom_kcp_config.stream = true;
            custom_kcp_config.nodelay = true;
            // 一定的延迟
            custom_kcp_config.dead_link_thres = 200;
        } else {
            // performance speed up for http sender other tunnel
            log::warn!("set tunnel config nodelay=true");
            custom_kcp_config.nodelay = true;
            custom_kcp_config.dead_link_thres = 100;
        }

        // channel send remote auth message
        sender_tx
            .send((
                remote_client_id.to_owned(),
                TunnelSignalIceMsg::RemoteAuth(IceAuth {
                    is_controlling,
                    ufrag: local_ufrag.clone(),
                    pwd: local_pwd.clone(),
                    auth_local_client_id: local_client_id.to_owned(),
                    auth_remote_client_id: remote_client_id.to_owned(),
                    tunnel_id: my_tunnel_id.to_owned(),
                    ekho_config: Some(custom_kcp_config), //default Config
                }),
            ))
            .await?;

        log::info!("new_tunnel_connect  get_local_user_credentials send");
        //create on_candidate callback
        let sender = sender_tx.clone();

        let can_remote_client_id = remote_client_id.to_owned();
        let can_local_client_id = local_client_id.to_owned();
        let can_tunnel_id = my_tunnel_id.to_owned();

        ice_agent.on_candidate(Box::new(
            move |c: Option<Arc<dyn Candidate + Send + Sync>>| {
                let can_remote_client_id = can_remote_client_id.to_owned();
                let can_local_client_id = can_local_client_id.to_owned();
                let can_tunnel_id = can_tunnel_id.to_owned();
                let mut sender = sender.clone();
                Box::pin(async move {
                    if let Some(c) = c {
                        let candidate = c.marshal();
                        log::info!("new_tunnel_connect  on_candidate: {}", &candidate);
                        //send local sdp candidate to remote
                        if let Err(e) = sender
                            .send((
                                can_remote_client_id.to_owned(),
                                TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
                                    candidate: candidate,
                                    can_local_client_id: can_local_client_id.to_owned(),
                                    can_remote_client_id: can_remote_client_id.to_owned(),
                                    is_controlling,
                                    tunnel_id: can_tunnel_id.to_owned(),
                                }),
                            ))
                            .await
                        {
                            log::error!("new_tunnel_connect on_candidate send err: {}", &e);
                        }
                    }
                })
            },
        ));
        let pair_change_tx = callback_tx.clone();
        // create agent callback.
        let pair_local_client_id = local_client_id.to_owned();
        let pair_remote_client_id = remote_client_id.to_owned();
        let pair_tunnel_id = my_tunnel_id.to_owned();
        ice_agent.on_selected_candidate_pair_change(Box::new(move |a, b| {
            log::info!(
                "new_tunnel_connect ICE on_selected_candidate_pair_change: {} {}",
                &a,
                &b
            );
            let mut pair_change_tx = pair_change_tx.clone();
            let mine_pair = candidate_to_json(a.clone());
            let remote_pair = candidate_to_json(b.clone());
            let pair_local_client_id = pair_local_client_id.to_owned();
            let pair_remote_client_id = pair_remote_client_id.to_owned();
            let pair_tunnel_id = pair_tunnel_id.to_owned();

            Box::pin(async move {
                if let Err(e) = pair_change_tx
                    .send(TunnelConnectMsg::OnTunnelPairChange(TunnelPair {
                        pair_local_client_id: pair_local_client_id.to_owned(),
                        pair_remote_client_id: pair_remote_client_id.to_owned(),
                        tunnel_id: pair_tunnel_id.to_owned(),
                        remote_pair: remote_pair,
                        mine_pair: mine_pair,
                    }))
                    .await
                {
                    log::error!(
                        "new_tunnel_connect on_selected_candidate_pair_change send err: {}",
                        &e
                    );
                }
            })
        }));

        let state_change_tx = callback_tx.clone();

        let sa_local_client_id = local_client_id.to_owned();
        let sa_remote_client_id = remote_client_id.to_owned();
        let sa_tunnel_id = my_tunnel_id.to_owned();
        //create connection state callback
        ice_agent.on_connection_state_change(Box::new(move |state: ConnectionState| {
            let mut state_change_tx = state_change_tx.clone();

            let sa_local_client_id = sa_local_client_id.to_owned();
            let sa_remote_client_id = sa_remote_client_id.to_owned();
            let sa_tunnel_id = sa_tunnel_id.to_owned();
            Box::pin(async move {
                log::info!(
                    "new_tunnel_connect ICE Connection State has changed: {}",
                    &state
                );
                if let Err(e) = state_change_tx
                    .send(TunnelConnectMsg::OnTunnelStateChange(TunnelStatus {
                        tunnel_id: sa_tunnel_id.to_owned(),
                        sa_local_client_id: sa_local_client_id.to_owned(),
                        sa_remote_client_id: sa_remote_client_id.to_owned(),
                        state,
                        msg: state.to_string(),
                    }))
                    .await
                {
                    log::error!(
                        "new_tunnel_connect on_connection_state_change send err: {}",
                        &e
                    );
                }
            })
        }));
        log::info!("new_tunnel_connect gather_candidates");
        // gather_candidates.
        ice_agent.gather_candidates()?;

        log::info!("new_tunnel_connect Connecting...");
        let has_connect = Arc::new(RwLock::new(false));

        let local_iscontrolling = is_controlling;
        // loop wait receive_rx received webrtc p2p connection from signal.
        loop {
            let mut loop_sender = sender_tx.clone();
            let mut cached_remote_ufrag: String = String::default();
            let mut cached_remote_pwd: String = String::default();

            let l_client_id = local_client_id.to_owned();
            let r_client_id = remote_client_id.to_owned();
            let t_id = my_tunnel_id.to_owned();

            let local_ufrag = local_ufrag.to_owned();
            let local_pwd = local_pwd.to_owned();
            // received message
            let item = if !*has_connect.read().await {
                //  first time  (! *has_connect) == true
                // wait for TunnelSignalIceMsg 15 seconds // turn default 25 seconds
                match tokio::time::timeout(Duration::from_secs(25), receive_rx.next()).await {
                    Ok(item) => match item {
                        Some(item) => item,
                        None => {
                            log::info!("sender_tx has gone break");
                            break;
                        }
                    },
                    Err(_e) => {
                        // if the future or stream does not complete in time, then it is canceled and an error is returned.
                        if *has_connect.read().await {
                            // first time  *has_connect = false
                            continue;
                        } else {
                            log::debug!("has_connect.read() _e = {}", _e);
                            if_no_kcp_set_err(
                                &l_client_id,
                                anyhow::anyhow!("tunnel connect timeout"),
                            )
                            .await;

                            if let Err(e) = callback_tx
                                .send(TunnelConnectMsg::OnTunnelStateChange(TunnelStatus {
                                    sa_local_client_id: l_client_id.to_owned(),
                                    sa_remote_client_id: r_client_id.to_owned(),
                                    tunnel_id: t_id.to_owned(),
                                    msg: "tunnel connect timeout".to_string(),
                                    state: ConnectionState::Failed,
                                }))
                                .await
                            {
                                log::error!("do_connect callback_tx timeout send err: {}", &e);
                            };

                            TunnelSignalIceMsg::Stop {
                                tunnel_id: t_id.to_owned(),
                                st_local_client_id: l_client_id.to_owned(),
                                st_remote_client_id: r_client_id.to_owned(),
                                kind: TunnelStopKind::ConnectTimout,
                            }
                        }
                    }
                }
            } else {
                match receive_rx.next().await {
                    Some(item) => item,
                    None => break,
                }
            };

            log::info!("new_tunnel_connect get msg {:?}", &item);
            match item {
                // received stop
                TunnelSignalIceMsg::Stop {
                    tunnel_id,
                    st_local_client_id: _,
                    st_remote_client_id: _,
                    kind: _,
                } => {
                    log::info!(
                        "new_tunnel_connect receive_rx Stop tunnel_id: {}",
                        tunnel_id
                    );
                    if tunnel_id == t_id {
                        log::warn!(
                            "new_tunnel_connect receive_rx tunnel_id equ stop: {}",
                            tunnel_id
                        );
                        close_tunnel_connect(&tunnel_id);
                        break;
                    } else {
                        log::warn!(
                            "new_tunnel_connect receive_rx tunnel_id not equ {} {}",
                            tunnel_id,
                            t_id
                        );
                    }
                    continue;
                }
                // received gather candidate
                TunnelSignalIceMsg::GatherCandidates { .. } => {
                    log::info!("new_tunnel_connect GatherCandidates");

                    if let Err(e) = ice_agent
                        .restart(local_ufrag.clone(), local_pwd.clone())
                        .await
                    {
                        log::warn!("GatherCandidates ice_agent.restart err: {:?}", e);
                    };
                    tracing::info!(%cached_remote_ufrag, %cached_remote_pwd, %local_ufrag, %local_pwd, "GatherCandidates");
                    if let Err(e) = ice_agent
                        .set_remote_credentials(
                            cached_remote_ufrag.clone(),
                            cached_remote_pwd.clone(),
                        )
                        .await
                    {
                        log::warn!(
                            "GatherCandidates ice_agent.set_remote_credentials err: {:?}",
                            e
                        );
                    }
                    if let Err(e) = ice_agent.gather_candidates() {
                        log::warn!("GatherCandidates ice_agent.gather_candidates err: {:?}", e);
                    }
                }
                TunnelSignalIceMsg::RemoteAuth(IceAuth {
                    is_controlling: remote_is_controlling,
                    ufrag,
                    pwd,
                    auth_local_client_id: _,
                    auth_remote_client_id,
                    tunnel_id,
                    ekho_config: remote_config,
                }) => {
                    if t_id != tunnel_id {
                        log::warn!(
                            "new_tunnel_connect receive_rx RemoteAuth tunnel_id not equ {} {}",
                            &t_id,
                            &tunnel_id
                        );
                        continue;
                    }
                    if l_client_id != auth_remote_client_id {
                        log::warn!(
                            "new_tunnel_connect receive_rx RemoteAuth remote_device_id not equ {} {}",
                            &l_client_id,
                            auth_remote_client_id,
                        );
                        continue;
                    }
                    let remote_ufrag = ufrag.clone();
                    let remote_pwd = pwd.clone();
                    let mut callback_tx = callback_tx.clone();
                    match ice_agent.get_local_candidates().await {
                        Ok(l_cans) => {
                            log::warn!(
                                "new_tunnel_connect RemoteAuth get_local_candidates ok,l_cans = {}",
                                l_cans.len()
                            );
                            for c in &l_cans {
                                let candidate = c.marshal();
                                //send local sdp candidate to remote
                                if let Err(e) = loop_sender
                                    .send((
                                        r_client_id.clone().to_owned(),
                                        TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
                                            candidate: candidate,
                                            can_local_client_id: l_client_id.clone().to_owned(),
                                            can_remote_client_id: r_client_id.to_owned(),
                                            is_controlling,
                                            tunnel_id: t_id.clone().to_owned(),
                                        }),
                                    ))
                                    .await
                                {
                                    log::error!("new_tunnel_connect on_candidate send err: {}", &e);
                                }
                            }
                            log::info!("ice_agent.get_local_candidates().await ok!");
                        }
                        Err(err) => {
                            log::info!("ice_agent.get_local_candidates().await err,{}", err);
                        }
                    }
                    let ice_agent2 = Arc::clone(&ice_agent);
                    let has_connect_inner = has_connect.clone();
                    // let local_config = local_config.clone();

                    let need_dail;
                    if remote_is_controlling {
                        // make remote dail and local accept
                        need_dail = false;
                    } else {
                        // make remote accept and local dail
                        need_dail = true;
                    }
                    tokio::spawn(async move {
                        let remote_ufrag = remote_ufrag.to_owned();
                        let remote_pwd = remote_pwd.to_owned();
                        // let local_ufrag = local_ufrag.to_owned();
                        // let local_pwd = local_pwd.to_owned();

                        let (cancel_tx, cancel_rx) = tokio::sync::mpsc::channel(1);
                        let conn: Arc<dyn Conn + Send + Sync> = if need_dail {
                            // if received remote auth info ,then dail from here
                            match ice_agent2
                                .dial(cancel_rx, remote_ufrag.clone(), remote_pwd.clone())
                                .await
                            {
                                Ok(conn) => {
                                    cached_remote_ufrag = remote_ufrag;
                                    cached_remote_pwd = remote_pwd;

                                    log::info!(
                                        "ice_agent2.dial remote_ufrag = {},remote_pwd = {}",
                                        cached_remote_ufrag,
                                        cached_remote_pwd
                                    );
                                    conn
                                }
                                Err(e) => {
                                    log::error!("do_connect ice_agent.dial err: {}", &e);
                                    return;
                                }
                            }
                        } else {
                            match ice_agent2
                                .accept(cancel_rx, remote_ufrag.clone(), remote_pwd.clone())
                                .await
                            {
                                Ok(conn) => {
                                    cached_remote_ufrag = remote_ufrag;
                                    cached_remote_pwd = remote_pwd;

                                    log::info!(
                                        "ice_agent2.accept remote_ufrag = {},remote_pwd = {}",
                                        cached_remote_ufrag,
                                        cached_remote_pwd
                                    );
                                    conn
                                }
                                Err(e) => {
                                    log::error!("do_connect ice_agent.accept err: {}", &e);
                                    return;
                                }
                            }
                        };
                        *has_connect_inner.write().await = true;
                        let connect_use_time = SystemTime::now()
                            .duration_since(start_time)
                            .unwrap_or_default()
                            .as_millis();
                        log::info!("tunnel connect_use_time: {}ms", connect_use_time);
                        // let ekho_config = Config::default();

                        if let Err(e) = callback_tx
                            .send(TunnelConnectMsg::OnTunnelConnectReady(TunnelConn {
                                ice_agent: ice_agent2,
                                conn,
                                remote_client_id: r_client_id.to_owned(),
                                local_client_id: l_client_id.to_owned(),
                                tunnel_id: t_id.to_owned(),
                                cancel_tx: cancel_tx,
                                is_controlling: local_iscontrolling,
                                ekho_config: remote_config.unwrap_or_default(),
                            }))
                            .await
                        {
                            log::error!("do_connect on_connection_state_change send err: {}", &e);
                        }
                    });
                }
                TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
                    candidate,
                    can_local_client_id: _,
                    can_remote_client_id,
                    tunnel_id,
                    is_controlling: _remote_is_controlling,
                }) => {
                    if t_id != tunnel_id {
                        log::warn!(
                            "new_tunnel_connect receive_rx RemoteCandidate tunnel_id not equ {} {}",
                            &t_id,
                            &tunnel_id
                        );
                        continue;
                    }
                    if l_client_id != can_remote_client_id {
                        log::warn!(
                            "new_tunnel_connect receive_rx RemoteCandidate remote_device_id not equ {} {}",
                            &l_client_id,
                            can_remote_client_id
                        );
                        continue;
                    }
                    if let Ok(c) = unmarshal_candidate(&candidate) {
                        log::info!(
                            "add_remote_candidate: {} is_controlling: {}",
                            &c,
                            is_controlling
                        );
                        let c: Arc<dyn Candidate + Send + Sync> = Arc::new(c);
                        let _ = ice_agent.add_remote_candidate(&c);
                    } else {
                        log::error!(
                            "new_tunnel_connect receive_rx RemoteCandidate unmarshal_remote_candidate err"
                        );
                    }
                }
            }
        }
        ice_agent.close().await?;
        log::info!("new_tunnel_connect end");
        Ok(())
    }
}

#[test]
fn test_uri_info() {
    let options = TunnelConfig::default();
    println!("test_uri_info uris: {:?}", options.turn_uri);
    println!(
        "test_uri_info: {:?}",
        options.turn_uri.to_webrtc_url().unwrap()
    );
}
